package com.wengcheng.pay.zookeeper;

import com.alibaba.dubbo.config.ProtocolConfig;
import com.alibaba.dubbo.config.ServiceConfig;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.SortedSet;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import javafx.collections.transformation.SortedList;
import org.I0Itec.zkclient.IZkChildListener;
import org.I0Itec.zkclient.IZkDataListener;
import org.I0Itec.zkclient.ZkClient;
import org.I0Itec.zkclient.exception.ZkNodeExistsException;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import sun.reflect.generics.tree.Tree;

/**
 * Created by adengdeng on 2017/8/22
 */
public class MasterSelector {

    private ZkClient zkClient;

    private final static String MASTER_PATH = "/master"; //需要争抢的节点
    private final static String PAY_CENTER = "/payCenter"; //需要争抢的节点

    private IZkDataListener dataListener; //注册节点内容变化
    private IZkChildListener childListener; //注册节点内容变化

    private PayCenter server;  //其他服务器
    private String serverPath;  //其他服务器

    private PayCenter master;  //master节点
    private String masterPath;  //master节点

    private boolean isRunning = false;

    ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(1);

    public MasterSelector(PayCenter server, ZkClient zkClient) {

        this.server = server;
        this.zkClient = zkClient;

        this.dataListener = new IZkDataListener() {

            public void handleDataChange(String s, Object o) throws Exception {
                System.out.println(Thread.currentThread().getName() + " [handleDataChange] " + s);

                //绑定节点数据变化监听
                zkClient.subscribeDataChanges(s, dataListener);
            }

            public void handleDataDeleted(String s) throws Exception {
                //节点如果被删除, 发起选主操作

                System.out.println(Thread.currentThread().getName() + " [handleDataDeleted] " + s);

                chooseMaster();
            }
        };

        this.childListener = new IZkChildListener() {
            @Override
            public void handleChildChange(String parentPath, List<String> currentChilds) throws Exception {

                System.out.println(Thread.currentThread().getName() + " [handleChildChange] " + parentPath + " [child]: " + currentChilds.toString());

                for (String child : currentChilds) {

                    //给每个子节点绑定节点监听
                    zkClient.subscribeDataChanges(PAY_CENTER + "/" + child, dataListener);
                }
            }
        };
    }

    public void start() throws InterruptedException {
        //开始选举
        if (!isRunning) {
            isRunning = true;
            if (!zkClient.exists(PAY_CENTER)) {
                zkClient.createPersistent(PAY_CENTER);
            }
            zkClient.subscribeChildChanges(PAY_CENTER, childListener); //注册节点事件

            //创建临时有序节点
            String path = zkClient.createEphemeralSequential(PAY_CENTER + MASTER_PATH, server);

            System.out.println(Thread.currentThread().getName() + " [创建临时有序节点] " + path);

            serverPath = path;

            chooseMaster();
        }
    }

    public void stop() {
        //停止
        if (isRunning) {
            isRunning = false;
            scheduledExecutorService.shutdown();
            zkClient.unsubscribeChildChanges(PAY_CENTER, childListener);
            releaseMaster();
        }
    }

    //具体选master的实现逻辑
    private void chooseMaster() throws InterruptedException {
        if (!isRunning) {
            System.out.println("当前服务没有启动");
            return;
        }


        List<String> childs = zkClient.getChildren(PAY_CENTER);

        Collections.sort(childs);

        if (childs.size() > 1) {

            String path = PAY_CENTER + "/" + childs.get(0);

            PayCenter first = zkClient.readData(path, true);

            if (first != null && server.getId() == first.getId()) {

                master = first;

                masterPath = path;

                System.out.println(Thread.currentThread().getName() + " master节点是：" + masterPath);

                //启动dubbo pay服务 保证下面删除节点的时候pay服务已经注册
                ClassPathXmlApplicationContext context =
                        new ClassPathXmlApplicationContext(new String[]{"META-INF/spring/dubbo-pay-provider.xml"});
                context.start();

                //定时器
                //master释放(master 出现故障）,没5秒钟释放一次
                scheduledExecutorService.schedule(() -> {

                    releaseMaster();//释放锁
                }, 2, TimeUnit.SECONDS);
            } else {
                cleanMasterRef();

                //监听其他节点变化
                for (int i = 1; i < childs.size(); i++) {

                    zkClient.subscribeDataChanges(PAY_CENTER + "/" + childs.get(i), dataListener);
                }
            }
        } else {

            cleanMasterRef();

            //在父节点上绑定子节点变化监听
            zkClient.subscribeChildChanges(PAY_CENTER, childListener);
        }
    }

    private void cleanMasterRef() {

        master = null;

        masterPath = null;
    }

    private void releaseMaster() {

        System.out.println(Thread.currentThread().getName() + " 执行releaseMaster:" + serverPath);
        //释放锁(故障模拟过程)
        //判断当前是不是master，只有master才需要释放
        if (checkIsMaster()) {
            zkClient.delete(serverPath); //删除

            //关闭所有dubbo服务
//            ProtocolConfig.destroyAll();

        }
    }

    private boolean checkIsMaster() {
        //判断当前的server是不是master
        PayCenter payCenter = zkClient.readData(masterPath);
        if (payCenter.getName().equals(server.getName())) {
            master = payCenter;
            return true;
        }
        return false;
    }

}
